import json

from pyspark import SparkConf, SparkContext

import defs_19

import os
os.environ['HADOOP_CONF_DIR'] = "/export/hadoop/bin/hadoop"

if __name__ == '__main__':
    conf = SparkConf().setAppName("test-yarn-1").setMaster("yarn")
    # 如果提交到集群运行，除了主代码以外，还依赖了其它的代码文件
    # 需要设置一个参数，来告知spark，还有依赖文件要同步上传到集群中
    # 参数叫做：spark.submit.pyFiles
    # 参数的值，可以是单个.py文件，也可以是.zip压缩包（有多个依赖文件的时候可以用zip压缩包上传）
    conf.set("spark.submit.pyFiles", "defs_19.py")
    sc = SparkContext(conf=conf)

    # 需求：读取data文件夹中的order.text文件，提取北京的数据，组合北京和商品类别进行输出同时对结果集进行去重，得到北京售卖的商品类别信息

    file_rdd = sc.textFile("hdfs://11.50.138.178:8020/tmp/hc/pydemo/data/input/order.json")

    # rdd split
    json_rdd = file_rdd.flatMap(lambda line: line.split("|"))

    # 通过 Python 内置的json库
    dict_rdd = json_rdd.map(lambda json_str: json.loads(json_str))

    print(dict_rdd.collect())

    bj_rdd = dict_rdd.filter(lambda d: d['areaName'] == '北京')

    # 组合背景 商品类型 形成新的字符串
    category_rdd = bj_rdd.map(lambda x: defs_19.city_with_category(x))

    # 对结果集进行去重操作
    result_rdd = category_rdd.distinct()

    # 输出
    print(result_rdd.collect())
